package org.ligson.spark.ml.statis;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaDoubleRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.DoubleFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.stat.MultivariateStatisticalSummary;
import org.apache.spark.mllib.stat.Statistics;
import org.ligson.spark.util.SparkContextUtil;

import java.util.Arrays;

public class StatisDemo {
    public static void main(String[] args) {
        JavaSparkContext context = SparkContextUtil.getLocalContext();
        JavaRDD<String[]> data = context.textFile("./src/main/resources/sample_stat.txt").map(new Function<String, String[]>() {
            public String[] call(String val) throws Exception {
                return val.split(" ");
            }
        });
        JavaRDD<double[]> data2 = data.map(new Function<String[], double[]>() {
            public double[] call(String[] v1) throws Exception {
                double[] doubles = new double[v1.length];
                for (int i = 0; i < v1.length; i++) {
                    doubles[i] = Double.parseDouble(v1[i]);
                }
                return doubles;
            }
        });
        JavaRDD<Vector> data3 = data2.map(new Function<double[], Vector>() {
            public Vector call(double[] v1) throws Exception {
                return Vectors.dense(v1);
            }
        });
        MultivariateStatisticalSummary summ = Statistics.colStats(data3.rdd());
        System.out.println(summ.max());
        System.out.println(summ.min());
        System.out.println(summ.mean());
        /*String result = data.reduce(new Function2<String, String, String>() {
            public String call(String v1, String v2) throws Exception {
                return v1 + ":" + v2;
            }
        });
        System.out.println(result);*/
        //JavaRDD<Vector> rdd = context.parallelize(Arrays.asList(new int[]{1, 2, 3, 4, 5, 5, 6, 7}));
        //Statistics.colStats(data.rdd());
    }
}
